package com.networkflow_analysis


import com.networkflow_analysis.bean.{UserBehavior, UvCount}
import org.apache.flink.api.common.functions.MapFunction
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.triggers.{Trigger, TriggerResult}
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
import redis.clients.jedis.Jedis

import scala.util.Random

/**
  * @Description: TODO QQ1667847363
  * @author: xiao kun tai
  * @date:2021/11/29 23:26
  *                  网站独立访客数
  *                  Uv统计，去重同用户,访问网站的总人数,未实现并发
  */
object UniqueVisitorWithBloomAdvance {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(8)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) //定义事件时间语义

    //从文件中读取数据，并转换成样例类,提取时间戳生成watermark
    //读取数据，转换成样例类提取时间戳和watermark
    val resource = getClass.getResource("/").toURI
    val filePath: String = "NetworkFlowAnalysis/src/main/resources/UserBehavior.csv"
    val fileStream: DataStream[String] = env.readTextFile(filePath)

    val dataStream: DataStream[UserBehavior] = fileStream.map(data => {
      val arr = data.split(",");
      UserBehavior(arr(0).toLong, arr(1).toLong, arr(2).toInt, arr(3), arr(4).toLong)
    })
      .assignAscendingTimestamps(_.timestamp * 1000L)

    val UvStream: DataStream[UvCount] = dataStream.filter(_.behavior == "pv")
      .map(new MyMapper)
      .keyBy(_._1)
      .timeWindow(Time.hours(1))
      .trigger(new MyTrigger) //自定义触发器
      .process(new UvCountWithBloom)

    UvStream.print("uv with bloom job")


    env.execute("pv job")
  }

  //自定义Mapper，随机生成分组的key
  class MyMapper extends MapFunction[UserBehavior, (String, Long)] {
    override def map(t: UserBehavior): (String, Long) = {
      (Random.nextString(10), t.userId)
    }
  }

  //自定义触发器，每来一条数据，直接触发窗口计算并清空窗口状态
  class MyTrigger extends Trigger[(String, Long), TimeWindow] {
    /**
      * CONTINUE ： 不进行操作，等待。
      * FIRE ： 触发计算且数据保留。
      * PRUGE ： 窗口内部数据清除且不触发计算
      * FIRE_AND_PURGE : 触发计算并清除对应的数据。
      */
    //每一个数据进入窗口都会触发
    override def onElement(t: (String, Long), l: Long, w: TimeWindow, triggerContext: Trigger.TriggerContext): TriggerResult = TriggerResult.FIRE_AND_PURGE

    //根据接入窗口的ProcessTime进行触发操作
    override def onProcessingTime(l: Long, w: TimeWindow, triggerContext: Trigger.TriggerContext): TriggerResult = TriggerResult.CONTINUE

    //根据接入窗口的EventTime进行触发操作
    override def onEventTime(l: Long, w: TimeWindow, triggerContext: Trigger.TriggerContext): TriggerResult = TriggerResult.CONTINUE

    //执行窗口及状态数据的清除方法
    override def clear(w: TimeWindow, triggerContext: Trigger.TriggerContext): Unit = {}
  }

  //自定义一个布隆过滤器,这要是一个位图和hash函数
  class Bloom(size: Long) extends Serializable {
    private val cap = size

    //hash函数
    def hash(value: String, seed: Int): Long = {
      var result = 0
      for (i <- 0 until value.length) {
        result = result * seed + value.charAt(i)
      }
      //返回hash值，要映射到cap范围内
      (cap - 1) & result
    }
  }

  //实现自定义的窗口处理函数
  class UvCountWithBloom extends ProcessWindowFunction[(String, Long), UvCount, String, TimeWindow] {
    //定义redis连接以及布隆过滤器
    lazy val jedis = new Jedis("192.168.88.106", 6379)
    lazy val bloomFilter = new Bloom(1 << 29) //位的个数：2^6(64) * 2^20(1M) * 2^3 (8bit)
    //本来是收集所有数据、窗口触发计算的时候才会调用，现在没来一条数据调用一次
    override def process(key: String, context: Context, elements: Iterable[(String, Long)], out: Collector[UvCount]): Unit = {
      //先定义redis中存储位图的key
      val storedBitMapKey = context.window.getEnd.toString
      //另外将当前窗口的uv count值，作为保存到redis里，用作叫做uvcount的hash表来保存（windowEnd,count）
      val uvCountMap: String = "uvcount"
      val currentKey = context.window.getEnd.toString
      var count = 0L
      //从redis中取出当前窗口的uv count值
      if (jedis.hget(uvCountMap, currentKey) != null) {
        count = jedis.hget(uvCountMap, currentKey).toLong
      }
      //去重：判断当前userId的hash值对应的位图位置，是否为0
      val userId = elements.last._2.toString
      //计算hash值，就对应位图中的偏移量
      val offset = bloomFilter.hash(userId, 61)
      //用redis的位操作命令，取bitmap中对应位的值
      val isExist = jedis.getbit(storedBitMapKey, offset)
      if (!isExist) {
        //如果不存在，那么位图对应位置置1，并且将count的值+1
        jedis.setbit(storedBitMapKey, offset, true)
        jedis.hset(uvCountMap, currentKey, (count + 1).toString)
      }
      out.collect(UvCount(storedBitMapKey.toLong, count))


    }
  }

}
